// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jshin

#include "coin2/exchange/gdax/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/gdax/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::gdax::feed_v1::impl {

void GdaxFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: " << GetParseError_En(document.GetParseError());
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }
  if (document.HasMember("type")) {
    const std::string& type = document["type"].GetString();
    const auto& channel = GdaxChannel::Parse(type);
    if (!CheckChannelFilter(channel.type)) {
      return;
    }

    if (!document.HasMember("product_id")) {
      LOG(INFO) << "[Gdax] unknown msg for " << packet.topic_name << ": " << packet.payload.data();
      return;
    }
    const std::string& native_symbol = document["product_id"].GetString();
    if (type == "match") {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }
      if (parser_config_.has_trade_sampling_frequency_bps()) {
        auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
        Trade _trade{};
        GdaxFeedUpdateParser::ParseTrade(document, &_trade);
        bool aggregated = slim_trade->TryAggregateSlimTrade(
                packet.timestamp,
                _trade.price,
                _trade.fill_qty,
                _trade.side,
                _trade.timestamp,
                parser_config_.trade_sampling_frequency_bps());
        int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
        if (!aggregated ||
            timespan > parser_config_.trade_sampling_time_length()) {
          auto* trade = upd_mgr_.GetTradeMutable();
          CHECK(slim_trade->PopulateTradeAndClear(trade));
          upd_mgr_.PublishTrade(native_symbol, false);
        }
      } else {
        Trade* trade = upd_mgr_.GetTradeMutable();
        GdaxFeedUpdateParser::ParseTrade(document, trade);
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else if (type == "l2update") {
      // Once market is busy, gdax may set invalid time
      const std::string& time_str = document["time"].GetString();
      if (time_str == "0001-01-01T00:00:00.000000Z") {
        LOG(INFO) << "[Gdax Feed Processor] invalid timestamp" << packet.topic_name << ": "
            << packet.payload.data();
        return;
      }

      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        if (parser_config_.has_sampling_frequency_bps()) {
          GdaxFeedUpdateParser::ParseBook(book, document, true);
          bool sampled = book->SampleByFrequency(
                  book->Bid0(),
                  book->Ask0(),
                  parser_config_.sampling_frequency_bps());
          if (parser_config_.has_trade_sampling_frequency_bps()) {
            auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
            int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
            if (sampled ||
                timespan > parser_config_.trade_sampling_time_length()) {
              auto* trade = upd_mgr_.GetTradeMutable();
              if (slim_trade->PopulateTradeAndClear(trade)) {
                upd_mgr_.PublishTrade(native_symbol, false);
              }
            }
          }
          if (sampled) {
            upd_mgr_.PublishBook();
          }
        } else if (publish_bbo_book_only_) {
          std::optional<PriceQty> bid_pre = book->Bid0();
          std::optional<PriceQty> ask_pre = book->Ask0();
          GdaxFeedUpdateParser::ParseBook(book, document, publish_bbo_book_only_);
          std::optional<PriceQty> bid_post = book->Bid0();
          std::optional<PriceQty> ask_post = book->Ask0();
          bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
          if (is_bbo_changed) {
            upd_mgr_.PublishBook();
          }
        } else {
          GdaxFeedUpdateParser::ParseBook(book, document, false);
          upd_mgr_.PublishBook();
        }
      }
    } else if (type == "snapshot") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        GdaxFeedUpdateParser::ParseSnapshot(book, document, packet.timestamp);
        upd_mgr_.PublishBook();
      }
    } else {
      LOG(INFO) << "[Gdax] unknown msg for " << packet.topic_name << ": " << packet.payload.data();
    }
  } else if (document.HasMember("url") && document["url"].GetString()) {
    const std::string& url = document["url"].GetString();
    if (url.find("candles") != std::string::npos) {
      std::vector<std::string> url_comp = Split(url, "?");
      CHECK_EQ(url_comp.size(), 2);
      std::vector<std::string> url_paths = Split(url_comp[0], "/");
      CHECK_GT(url_paths.size(), 3);
      const std::string& native_symbol = url_paths[url_paths.size() - 2];

      auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
      const std::string& kline_interval = parameters["granularity"];
      auto interval = GdaxFeedUpdateParser::ParseKlineInterval(kline_interval);
      auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (klines != nullptr) {
        // query is split up several requests
        int page = -1;
        if (parameters.count("page") > 0) {
          page = std::stod(parameters["page"]);
        }
        if (page == 0) {
          klines->SnapshotClear();
        } else {
          klines->UpdateClear();
        }
        bool to_publish = false;
        const auto& data = document["data"].GetArray();
        for (int i = data.Size() - 1; i >= 0; i--) {
          auto& kline_data = data[i];
          bool freezed = GdaxFeedUpdateParser::ParseKline(kline_data,
              klines, interval, packet.timestamp);
          to_publish = to_publish || freezed;
        }
        if (to_publish) upd_mgr_.PublishKline(native_symbol);
      }
    } else {
      LOG(INFO) << "[Gdax Feed Processor] unkown rest msg for url " << url;
    }
  } else {
    // meta msg like subscription response...
    LOG(INFO) << "[Gdax Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}

}  // namespace coin2::exchange::gdax::feed_v1::impl
